package com.atguigu.gmall.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.app.BaseAppV2;
import com.atguigu.gmall.bean.OrderDetail;
import com.atguigu.gmall.bean.OrderInfo;
import com.atguigu.gmall.bean.OrderWide;
import com.atguigu.gmall.common.Constant;
import com.atguigu.gmall.util.DimUtil;
import com.atguigu.gmall.util.JdbcUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.time.Duration;
import java.util.Map;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/7/31 14:05
 */
public class DwmOrderWideApp extends BaseAppV2 {
    public static void main(String[] args) {
        new DwmOrderWideApp().init(
            3003,
            1,
            "DwmOrderWideApp", "DwmOrderWideApp",
            Constant.TOPIC_DWD_ORDER_INFO, Constant.TOPIC_DWD_ORDER_DETAIL
        );
    }
    
    @Override
    protected void run(StreamExecutionEnvironment env, Map<String, DataStreamSource<String>> streams) {
        // 1. 事实表进行join
        SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDim = factJoin(streams);
        // 2. join的维度数据
        SingleOutputStreamOperator<OrderWide> orderWideStreamWithDim = dimJoin(orderWideStreamWithoutDim);
        orderWideStreamWithDim.print();
        
        // 3. 把宽表写入到dwm层(kafka)
    }
    
    private SingleOutputStreamOperator<OrderWide> dimJoin(SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDim) {
        /*
        join维度:
            每一个OrderWide都去hbase里查询相应的维度
         */
        return orderWideStreamWithoutDim.map(new RichMapFunction<OrderWide, OrderWide>() {
            
            private Connection phoenixConn;
            
            @Override
            public void open(Configuration parameters) throws Exception {
                phoenixConn = JdbcUtil.getPhoenixConnection();
            }
            
            @Override
            public OrderWide map(OrderWide wide) throws Exception {
                // 1. 补充 dim_user_info  select * from t where id=?
                JSONObject userInfo = DimUtil.readDimFromPhoenix(phoenixConn, Constant.DIM_USER_INFO, wide.getUser_id().toString());
                wide.setUser_gender(userInfo.getString("GENDER"));
                wide.setUser_age(userInfo.getString("BIRTHDAY"));
                
                // 2. 省份
                JSONObject baseProvince = DimUtil.readDimFromPhoenix(phoenixConn, Constant.DIM_BASE_PROVINCE, wide.getProvince_id().toString());
                wide.setProvince_3166_2_code(baseProvince.getString("ISO_3166_2"));
                wide.setProvince_area_code(baseProvince.getString("AREA_CODE"));
                wide.setProvince_iso_code(baseProvince.getString("ISO_CODE"));
                wide.setProvince_name(baseProvince.getString("NAME"));
                
                // 3. sku
                JSONObject skuInfo = DimUtil.readDimFromPhoenix(phoenixConn, Constant.DIM_SKU_INFO, wide.getSku_id().toString());
                wide.setSku_name(skuInfo.getString("SKU_NAME"));
    
                wide.setSpu_id(skuInfo.getLong("SPU_ID"));
                wide.setTm_id(skuInfo.getLong("TM_ID"));
                wide.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));
                
                // 4. spu
                JSONObject spuInfo = DimUtil.readDimFromPhoenix(phoenixConn, Constant.DIM_SPU_INFO, wide.getSpu_id().toString());
                wide.setSpu_name(spuInfo.getString("SPU_NAME"));
                // 5. tm
                JSONObject tmInfo = DimUtil.readDimFromPhoenix(phoenixConn, Constant.DIM_BASE_TRADEMARK, wide.getTm_id().toString());
                wide.setTm_name(tmInfo.getString("TM_NAME"));
                
                // 5. c3
                JSONObject c3Info = DimUtil.readDimFromPhoenix(phoenixConn, Constant.DIM_BASE_CATEGORY3, wide.getCategory3_id().toString());
                wide.setCategory3_name(c3Info.getString("NAME"));
                
                return wide;
            }
        });
        
    }
    
    private SingleOutputStreamOperator<OrderWide> factJoin(Map<String, DataStreamSource<String>> streams) {
        KeyedStream<OrderInfo, Long> orderInfoStream = streams
            .get(Constant.TOPIC_DWD_ORDER_INFO)
            .map(info -> JSON.parseObject(info, OrderInfo.class))
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((info, ts) -> info.getCreate_ts())
            )
            .keyBy(OrderInfo::getId);
        
        KeyedStream<OrderDetail, Long> orderDetailStream = streams
            .get(Constant.TOPIC_DWD_ORDER_DETAIL)
            .map(info -> JSON.parseObject(info, OrderDetail.class))
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((detail, ts) -> detail.getCreate_ts())
            )
            .keyBy(OrderDetail::getOrder_id);
        
        return orderInfoStream
            .intervalJoin(orderDetailStream)
            .between(Time.seconds(-5), Time.seconds(5))
            .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                @Override
                public void processElement(OrderInfo left,
                                           OrderDetail right,
                                           Context ctx,
                                           Collector<OrderWide> out) throws Exception {
                    out.collect(new OrderWide(left, right));
                }
            });
        
    }
}
